


# 数据重塑和汇总工具 类似excl中的数据透视表

import pandas as pd
import numpy as np

df = pd.DataFrame({
    '日期': ['2023-01-01', '2023-01-01', '2023-01-02', '2023-01-02'],
    '产品': ['A', 'B', 'A', 'B'],
    '城市': ['北京', '上海', '北京', '上海'],
    '销量': [100, 150, 120, 180],
    '价格': [10, 15, 12, 16]
})


# 自定义函数
def cus_agg(x):
    return x.max() - x.min()

cus_aggfunc = {
    "销量":"sum",
    "价格": ["mean", np.std]
}

result = pd.pivot_table(
    df, values=["销量"], index=["产品"], columns=["日期"], 
    aggfunc='sum', fill_value=0, # fillna() 提前处理缺失值，不要在这一部处理提升效率
    margins=True, margins_name="总计", sort=False # 关闭排序减少计算开销
)
# aggfunc=cus_agg

print(result)

query_result = result.query("日期"=="2023-01-02")



pivot = pd.pivot_table(df, index='产品', values=['收入', '成本'], aggfunc='sum')
pivot['利润'] = pivot['收入'] - pivot['成本']
pivot['利润率'] = pivot['利润'] / pivot['收入']
print(pivot)


result = pd.pivot_table(
    df,
    values=['销量', '价格'],
    index=['日期', '城市'],
    columns='产品',
    aggfunc={'销量': 'sum', '价格': 'mean'},
    fill_value=0,
    margins=True,
    margins_name='总计'
)

print(result)
#                  价格                  销量
# 产品                A     B     总计      A      B   总计
# 日期         城市
# 2023-01-01 上海   0.0  15.0  15.00    0.0  150.0  150
#            北京  10.0   0.0  10.00  100.0    0.0  100
# 2023-01-02 上海   0.0  16.0  16.00    0.0  180.0  180
#            北京  12.0   0.0  12.00  120.0    0.0  120
# 总计             11.0  15.5  13.25  220.0  330.0  550


# 总计 列 降序 排序
result.sort_values(by='总计', ascending=False)
# 多列排序：部门升序，季度销售额降序
result.reset_index().sort_values(['部门', '季度销售额'], ascending=[True, False])
# 对多级索引的透视表，先重置索引再排序
result.reset_index().sort_values(['日期', '城市', '销量'], ascending=[True, True, False])
# 北京地区的销售
result.query("城市 == '北京'")
# 销量 >100 且利润率为正
result.query("销量 > 100 & 利润率 > 0.1")
# 如部门名称含“销售”
result[result.index.str.contains('销售')]
# 3. 筛选低绩效部门（季度销售额<均值）
mean_sales = pivot.mean().mean()  # 计算全局均值
low_perf = pivot.query('总计 < @mean_sales')  # 动态引用变量
# 跨层级排序：
# 需先 reset_index() 将索引转为列，排序后再 set_index() 还原结构

# Top N ： sort_values(by='汇总列').head(N)






# pivot_table 本质上是 groupby 的特例

# groupby更适合需要灵活分组后操作（如过滤、转换）或复杂计算的场景
# pivot_table则擅长快速生成业务报表、多维交叉对比等结构化输出需求

# 特性	            pivot_table	                        groupby
# 输出结构	    二维表格（行+列维度）	               一维分组对象
# 多维度分析	    支持行列双向分类	                     仅支持行分组
# 聚合灵活性	    支持多列不同聚合函数	                  需结合 agg 实现
# 可视化友好度	直接生成表格，适合报表输出	              需手动重塑数据


# groupby：采用“拆分-应用-合并”机制，生成分层索引的 Series/DataFrame。内存效率更高（仅存储分组键和聚合结果），尤其适合大数据集的分步处理。
# pivot_table：输出二维表格结构，需为所有行列组合分配内存。当维度组合基数高时（如城市×产品×日期），
#               可能因生成大量空值而内存激增,存储行列笛卡尔积，内存易爆炸


# groupby 在链式操作（如 filter() → transform()）中更高效，避免生成中间表格
# pivot_table 在多维度交叉分析（如行+列+值）时优化更好，因内部复用分组结果

# groupby 结果需 unstack() 或 reset_index() 转为二维表格，额外增加计算开销。
# pivot_table 直接输出结构化表格，省去转换步骤，但牺牲了分组对象的灵活性

# 数据类型转换：将分组键转为 category 类型，减少内存占用


# ---------------------------分块处理数据-----------------------------------
chunks = pd.read_csv("data.csv", chunksize=1e6)
results = [chunk.groupby('key')['value'].sum() for chunk in chunks]
final_result = sum(results)



# 并行计算：使用Dask或Modin库，将任务分布到多核/集群
import dask.dataframe as dd
ddf = dd.read_csv("data.csv")
pivot = ddf.pivot_table(index='region', columns='product', values='sales', aggfunc='sum')
# 列式存储：将数据转为Parquet格式，提升I/O效率


# 聚合函数优化：用sum代替mean（省去计数步骤），或预计算聚合列
# 速度提升：sum + count 比直接 mean 快 30%~50%（尤其在大数据场景）；
# 内存节省：避免生成中间结果，减少内存峰值










# Groupby 的核心原理：拆分-应用-合并
# groupby 的本质是将数据集按指定条件拆分为子集，对每个子集独立执行操作，最后合并结果


import pandas as pd
import numpy as np

# 创建示例数据集
data = {
    'Store': ['A', 'B', 'A', 'B', 'A', 'B'],
    'Product': ['Apple', 'Banana', 'Apple', 'Banana', 'Orange', 'Orange'],
    'Sales': [150, 200, 300, 150, 400, 250],
    'Cost': [50, 60, 70, 80, 90, 100]
}
df = pd.DataFrame(data)

# 基础分组：按Store列拆分
grouped = df.groupby('Store')  # 此时生成GroupBy对象，未实际计算
print(type(grouped))  # <class 'pandas.core.groupby.generic.DataFrameGroupBy'>

# 计算每家店的总销售额
store_sales = grouped['Sales'].sum()
print(store_sales)


# 按Store和Product分组，计算组内销售额最大值
multi_group = df.groupby(['Store', 'Product'])['Sales'].max()
print(multi_group)

#  单列多聚合
result = (
    df.groupby('Store')['Sales'].agg(['sum', 'mean', 'std'])  # 同时计算总和、均值、标准差
)
print(result)



# 多列差异化聚合
result = df.groupby('Store').agg(
    total_sales=('Sales', 'sum'),          # 对Sales列求和
    avg_cost=('Cost', lambda x: x.mean()), # 对Cost列用自定义函数
    max_sales=('Sales', max)               # 对Sales列求最大值
)
print(result)


# 命名聚合
result = (
    df.groupby("Store")
    .agg(
        min_cost=pd.NamedAgg(column="Cost", aggfunc="min"),
        sales_range=pd.NamedAgg(column="Sales", aggfunc=lambda x: x.max() - x.min())
    )
)
print(result)

# 多列差异化聚合 命名聚合 两种等价
# pd.NamedAgg 的优势： 支持附加参数（如分位数计算）
# 元组写法仅支持 (列名, 聚合函数) 的固定格式，无法传递额外参数（如 q）
q25 = pd.NamedAgg(column="Cost", aggfunc="quantile", q=0.25)





# 转换操作返回与原数据相同大小的结果，常用于数据标准化或创建新列
# 计算每个分组的成本均值，并扩展填充到原数据行
df['Group_Cost_Mean'] = df.groupby('Store')['Cost'].transform('mean')

# 标准化组内销售额：(当前值 - 组均值) / 组标准差
df['Sales_Zscore'] = (
    df.groupby('Store')['Sales'].transform(lambda x: (x - x.mean()) / x.std())
)
print(df)





# 过滤操作 (Filtration)：基于分组统计结果筛选子集，类似 SQL 的 HAVING 子句。
# 只保留总销售额超过700的组
filtered = df.groupby('Store').filter(lambda group: group['Sales'].sum() > 700)
print(filtered)
filtered = df.groupby('User').filter(lambda group: len(group) >= 2) # 保留组内行数≥2的分组（过滤掉组C）


# 处理缺失值分组
df.loc[6] = [None, 'Apple', 180, 55]  # 添加缺失值
result = df.groupby('Store', dropna=False)['Sales'].sum()  # 保留NaN组
print(result)


data = {'Category': ['X', 'X', 'Y', 'Y', 'Z'], 'Value': [5, 15, 10, 20, 25]}
df = pd.DataFrame(data)

# 过滤：组内不含值=5 且 组均值≥15
filtered = df.groupby('Category').filter(
    lambda g: (g['Value'] != 5).all() and g['Value'].mean() >= 15
)
print(filtered)  # 保留组Y（均值=15）和组Z（均值=25）
# (g['Value'] != 5).all() 确保组内无5，mean()>=15 要求均值达标

# --------------- 时间序列分组过滤 ---------------------------
df = pd.DataFrame({
    'Date': pd.date_range('2023-01-01', periods=6, freq='M'),
    'Sales': [200, 150, 300, 0, 250, 100]
})

# 按季度分组，保留组内总销售额>500的季度
filtered = df.groupby(pd.Grouper(key='Date', freq='Q')).filter(
    lambda g: g['Sales'].sum() > 500
)
print(filtered)

# --------------分类数据过滤-----------------
# 保留组内包含类型'Y'的分组
filtered = df.groupby('Group').filter(lambda g: 'Y' in g['Type'].values)






# 获取特定分组数据
group_a = df.groupby('Store').get_group('A')  # 提取Store=A的所有行



# 时间序列分组  
dates = pd.date_range('2023-01-01', periods=100, freq='D')
ts_df = pd.DataFrame({
    'date': dates,
    'value': np.random.randn(100)
})

# 按月份聚合
ts_df.groupby(ts_df['date'].dt.month)['value'].mean()



# 避免在大型数据集上用 apply：transform 或 agg 通常比 apply 快 10 倍以上
# 使用 Cython 优化函数：如 np.sum 比 Python 原生的 sum() 更快
# 关闭索引创建：as_index=False 可减少内存占用



# 方法	                操作粒度	           返回结果大小	                  典型场景
# agg	                 组级别	                    每组一行	            统计摘要 (sum/mean)
# transform	        元素级别	             与原数据相同	             数据标准化/填充
# apply	             组级别	                   任意形状	                   复杂业务逻辑



# apply的灵活应用：返回每组的前两行
def get_first_two(rows):
    return rows.head(2)
    
print(df.groupby('Store').apply(get_first_two))






# 复杂分组聚合
df.groupby(['category', pd.cut(df['price'], bins=[0, 10, 100])]).size()


















# filter vs where：filter 丢弃整个组，where 保留组结构但替换不满足条件的值为NaN
# 链式操作：可结合 .groupby().apply().filter() 实现多阶段处理，但需注意执行效率